package com.iot.service.mqtt.service.rel;

import com.iot.service.mqtt.model.request.TlMqttPubRelReq;
import com.iot.service.mqtt.service.StoreStrategy;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author: hszhou
 * @Date: 2025/4/18 14:46
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@Service
@ConditionalOnProperty(prefix = "mqtt", name = "store-strategy", havingValue = StoreStrategy.LOCAL)
public class LocalPubrelStoreServiceImpl implements IPubrelStoreService {

    public static final ConcurrentHashMap<String, ConcurrentHashMap<String, TlMqttPubRelReq>> PUBREL_MAP = new ConcurrentHashMap<>();

    @Override
    public Mono<Boolean> savePubRelMessage(String clientId, Long messageId, TlMqttPubRelReq req) {


        return Mono.fromSupplier(()->
            PUBREL_MAP.compute(clientId, (k, v) -> {
                if (v == null) {
                    v = new ConcurrentHashMap<>(16);
                }
                v.computeIfAbsent(String.valueOf(messageId), key -> req);
                return v;
            })).thenReturn(true);
    }

    @Override
    public Mono<TlMqttPubRelReq> clearPubRelMessage(String clientId, Long messageId) {
        return Mono.fromSupplier(()->  PUBREL_MAP.getOrDefault(clientId, new ConcurrentHashMap<>(16))
                .remove(String.valueOf(messageId)));
    }

    @Override
    public Mono<Boolean> clearPubRelMessage(String clientId) {
        return Mono.fromSupplier(()->  PUBREL_MAP.remove(clientId)).thenReturn(true);
    }

    @Override
    public Mono<TlMqttPubRelReq> findPubRelMessage(String clientId, Long messageId) {
        return Mono.fromSupplier(()->PUBREL_MAP.getOrDefault(clientId, new ConcurrentHashMap<>(16)).get(String.valueOf(messageId)));
    }

    @Override
    public Flux<TlMqttPubRelReq> findPubRelMessage(String clientId) {

       return Flux.create(fluxSink -> PUBREL_MAP.getOrDefault(clientId, new ConcurrentHashMap<>(16)).forEach((k, v) -> fluxSink.next(v)));

    }
}
